package com.bw.test2;

import org.apache.flink.types.Row;

import com.alibaba.alink.operator.batch.BatchOperator;
import com.alibaba.alink.operator.batch.dataproc.ImputerPredictBatchOp;
import com.alibaba.alink.operator.batch.dataproc.ImputerTrainBatchOp;
import com.alibaba.alink.operator.batch.source.MemSourceBatchOp;
import com.alibaba.alink.operator.stream.StreamOperator;
import com.alibaba.alink.operator.stream.dataproc.ImputerPredictStreamOp;
import com.alibaba.alink.operator.stream.source.MemSourceStreamOp;
import org.junit.Test;

import java.util.Arrays;
import java.util.List;

public class ImputerTrainBatchOpTest {
	@Test
	public void testImputerTrainBatchOp() throws Exception {
		// 数据集
		List <Row> df_data = Arrays.asList(
				Row.of(2, 1, 1),
				Row.of(3, 2, 1),
				Row.of(4, 3, 2),
				Row.of(2, 4, 1),
				Row.of(2, 2, 1),
				Row.of(4, 3, 2),
				Row.of(1, null, null)
		);

		BatchOperator <?> inOp = new MemSourceBatchOp(df_data, "col1 int, col2 int, col3 int");


		String[] selectedColNames = new String[] {"col2", "col3"};

		// 创建缺少值
		BatchOperator <?> trainOp = new ImputerTrainBatchOp()
			.setSelectedCols(selectedColNames).setStrategy("VALUE").setFillValue("3");


		// 缺失值模型跟数据集关联起来
		BatchOperator model = trainOp.linkFrom(inOp);


		// 预测
		BatchOperator <?> predictOp = new ImputerPredictBatchOp();


		predictOp.linkFrom(model, inOp).print();


	}
}